package client;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.log4j.Logger;

import common.messages.KVMessage;
import common.messages.KVMessageImpl;
import common.messages.MD5Hash;
import common.messages.MetaData;
import communicationLogic.ClientSocket;

public class KVStore implements KVCommInterface {
    private Logger logger = Logger.getRootLogger();
    private KVMessage ans;
    private ClientSocket clientSocket;
    private List<MetaData> metadata;

    private static final String TAIL = "ffffffffffffffffffffffffffffffff";
    private static final String HEAD = "00000000000000000000000000000000";

    /**
     * Initialize KVStore with address and port of KVServer
     * 
     * @param address
     *            the address of the KVServer
     * @param port
     *            the port of the KVServer
     */
    public KVStore(String address, int port) {
        try {
            clientSocket = new ClientSocket(address, port);
            this.metadata = new ArrayList<MetaData>();
            logger.info("ClientSocket established");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * Establishes the connection to the KV Server.
     * 
     * @throws Exception
     *             if connection could not be established.
     */
    @Override
    public void connect() throws Exception {
        // TODO Auto-generated method stub
        clientSocket.connect();
        logger.info("Connection established");
    }

    /**
     * disconnects the client from the currently connected server.
     */
    @Override
    public void disconnect() {
        // TODO Auto-generated method stub
        try {
            clientSocket.disConnect();
            logger.info("Connection closed");
        } catch (Exception e) {
        }
    }

    /**
     * Inserts a key-value pair into the KVServer.
     * 
     * @param key
     *            the key that identifies the given value.
     * @param value
     *            the value that is indexed by the given key.
     * @return a message that confirms the insertion of the tuple or an error.
     * @throws Exception
     *             if put command cannot be executed (e.g. not connected to any
     *             KV server).
     */
    @Override
    public synchronized KVMessage put(String key, String value)
            throws Exception {
        KVMessageImpl kvm = new KVMessageImpl(key, value, "PUT", "");
        byte[] message = kvm.toByteArray(kvm);
        clientSocket.send(message);
        logger.info("Message sent" + message.toString());
        byte[] answer = clientSocket.receive();
        this.setAns(KVMessageImpl.toKVMessage(answer));
        logger.info("Answer received" + answer.toString());
        // if client connect to wrong server, automatically connect to correct
        // server and resend the message
        if (this.ans.getStatus().toString()
                .equalsIgnoreCase("SERVER_NOT_RESPONSIBLE")) {
            updateMeta(this.ans.getMetadata());
            String[] correctAdd = route2Server(this.ans.getMetadata(), key)
                    .split(":");
            String newIP = correctAdd[0];
            int newPort = Integer.valueOf(correctAdd[1]);
            KVStore kvs = new KVStore(newIP, newPort);
            kvs.connect();
            // logger.warn("Currenct server is not responsible. Automatically connect to "
            // + newIP + ":" + newPort);
            // System.out.println("Currenct server is not responsible. Automatically connect to "
            // + newIP + ":" + newPort);
            // logger.info("resend the message");
            KVMessage outKvMessage = kvs.put(key, value);
            kvs.disconnect();
            return outKvMessage;

        }
        return ans;
    }

    /**
     * Retrieves the value for a given key from the KVServer.
     * 
     * @param key
     *            the key that identifies the value.
     * @return the value, which is indexed by the given key.
     * @throws Exception
     *             if put command cannot be executed (e.g. not connected to any
     *             KV server).
     */

    @Override
    public synchronized KVMessage get(String key) throws Exception {
        KVMessageImpl kvm = new KVMessageImpl(key, "", "GET", "");
        byte[] message = kvm.toByteArray(kvm);
        clientSocket.send(message);
        logger.info("Message sent" + message.toString());
        byte[] answer = clientSocket.receive();
        this.setAns(KVMessageImpl.toKVMessage(answer));
        logger.info("Answer received" + answer.toString());
        // if client connect to wrong server, automatically connect to correct
        // server and resend the message
        if (this.ans.getStatus().toString()
                .equalsIgnoreCase("SERVER_NOT_RESPONSIBLE")) {
            updateMeta(this.ans.getMetadata());
            String[] correctAdd = route2Server(this.ans.getMetadata(), key)
                    .split(":");
            String newIP = correctAdd[0];
            int newPort = Integer.valueOf(correctAdd[1]);
            KVStore kvs = new KVStore(newIP, newPort);
            kvs.connect();
            logger.warn("Currenct server is not responsible. Automatically connect to "
                    + newIP + ":" + newPort);
            System.out
                    .println("Currenct server is not responsible. Automatically connect to "
                            + newIP + ":" + newPort);
            logger.info("resend the message");
            KVMessage outKvMessage = kvs.get(key);
            kvs.disconnect();
            return outKvMessage;
        }
        return ans;
    }

    public KVMessage getAns() {
        return ans;
    }

    public void setAns(KVMessage ans) {
        this.ans = ans;
    }

    public ClientSocket getClientSocket() {
        return this.clientSocket;
    }

    public List<MetaData> getMetadata() {
        return metadata;
    }

    /**
     * find the correct server according to the metadata and hash value of the
     * key
     * 
     * @param meta
     * @param key
     * @return ip+port of new server
     */
    private String route2Server(List<MetaData> meta, String key) {
        String hashKey = new MD5Hash().doHash(key);
        String correctAdd = "";
        for (MetaData s : meta) {
            if (s.getHead().compareToIgnoreCase(s.getTail()) < 0) {
                if (hashKey.compareToIgnoreCase(s.getHead()) > 0
                        && hashKey.compareToIgnoreCase(s.getTail()) < 0) {
                    correctAdd = s.getIp() + ":" + s.getPort();
                }
            } else {
                if ((hashKey.compareToIgnoreCase(s.getHead()) > 0 && hashKey
                        .compareToIgnoreCase(TAIL) < 0)
                        || (hashKey.compareToIgnoreCase(HEAD) > 0 && hashKey
                                .compareToIgnoreCase(s.getTail()) < 0)) {
                    correctAdd = s.getIp() + ":" + s.getPort();
                }
            }
        }
        return correctAdd;
    }

    /**
     * update the cache of metadata
     * 
     * @param meta
     */
    private void updateMeta(List<MetaData> meta) {
        this.metadata = meta;
    }

}
